//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// 
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Lesser General Public License for more details.
// 
// You should have received a copy of the GNU Lesser General Public License
// along with this program.  If not, see http://www.gnu.org/licenses/.
// 

//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with this program.  If not, see http://www.gnu.org/licenses/.
//

#include <omnetpp.h>
#include "Structs.h"
#include "UDPBasicApp.h"
#include "Messages_m.h"

class PeerApp : public UDPBasicApp {

	
	protected:
	int id;
	int myStreamID;
	IPvXAddress addr;
	std::string s_addr;
	IPvXAddress trackerAddr;
	std::string s_trackerAddr;

	std::vector<SubStream> uploads;
	unsigned maxUploads;
	bool stable;
	int nretries;
	simtime_t advTime, stableTime, startTime;
	
	int rank;
	
	bool geo;
	bool delayed;
	cTopology topo;
	std::vector<addressEntry> table;

	//Downloads
	std::vector<std::string> peerAddresses;
	std::vector<Candidate> peerList;
	unsigned int peerListIt;
	SubStream *downloads[MAX_SUBSTREAM];
	unsigned int subStreamCount;

	double byteCount;

	TrackTimer *ttimer;
	DataTimer *dtimer;

	protected:
		virtual void initialize(int);
		virtual void finish();
		virtual void handleMessage (cMessage *msg);
		
		std::string getModuleName(std::string addr);
		int countHops (std::string addr);
		void setNextPeer();
		void getAllPeerMetrics();
		
		void requestPeerList();
		void periodicSend();
		void trackingCheck();
		bool downloadCheck();
		
		Candidate *getNextPeer();
		std::string getNextPeerAddr();
		
		int getLastChunk();
		bool searchPeers();
		bool searchPeers(int);
		void fullSearchPeers();
		
		void processPing (Ping *);
		void processPong (Pong *);
		void processPeerList(PeerList *);
		void processRequest(StreamReq *);
		void disconnectPeer(PeerExit *);
		void processData (Data *);
		void closeConnection (int peerID, int streamID, int subStreamID, std::string addr, int cause, bool upload, bool download);

	public:
		PeerApp();
};

Define_Module(PeerApp);

PeerApp::PeerApp() {
	UDPBasicApp::UDPBasicApp();
}

void PeerApp::initialize(int stage) {

	// because of IPAddressResolver, we need to wait until interfaces are registered,
	// address auto-assignment takes place etc.
	if (stage!=3)
        	return;

	counter = 0;
	numSent = 0;
	numReceived = 0;
	byteCount = 0;
	WATCH(numSent);
	WATCH(numReceived);
	WATCH(byteCount);

	localPort = par("localPort");
	destPort = par("destPort");

	const char *destAddrs = par("destAddresses");
	cStringTokenizer tokenizer(destAddrs);
	const char *token;
	while ((token = tokenizer.nextToken())!=NULL)
        	destAddresses.push_back(IPAddressResolver().resolve(token));

	//if (destAddresses.empty())
        //	return;

	bindToPort(localPort);

	maxUploads = par("maxUploads");
	id = par("id");
	myStreamID = par("streamID");
	for (unsigned i = 0; i < MAX_SUBSTREAM; i++) {
		downloads[i] = NULL;
	}
	subStreamCount = 0;
	peerListIt = 0;
	stable = false;
	nretries = 0;
	advTime = 0;
	stableTime = 0;
	startTime = 0;
	//WATCH_VECTOR(peerAddresses);
	
	topo.extractByProperty("node");
	ev << getFullPath() << ": found " << topo.getNumNodes() << " IP nodes\n";
	
	for (int i = 0; i < topo.getNumNodes(); i++) {
		cModule *mod = topo.getNode(i)->getModule();
		addressEntry e;
		e.moduleName = mod->getFullPath();
		e.address = IPAddressResolver().resolve(mod->getFullPath().c_str()).str();
		table.push_back(e);
	}
	
	ev << getFullPath() << ": " << topo.getNumNodes() << " Nodes\n";
	topo.calculateUnweightedSingleShortestPathsTo( topo.getNodeFor(getParentModule()) );
	
	rank = (int) par ("rank");
	geo = (bool) par ("geoAware");
	delayed = (bool) par ("delayed");

	dtimer = new DataTimer();
	ttimer = new TrackTimer();
	
	cMessage *c = new cMessage("ConfigureMsg", CONFIGURE);
	scheduleAt(simTime() + par("startedAt"),c);
}

void PeerApp::finish() {
	unsigned subs;

	ev << "----------\n";
	if (!stable)
		for (subs = 0; subs < MAX_SUBSTREAM; subs++) {
			if (downloads[subs] != NULL)
				ev << getFullPath() << ": Substream " << subs << " received chunks: " << downloads[subs]->getNextChunk() << "\n";
			else
				ev << getFullPath() << ": Missing substream " << subs << "\n";
		}
	else
		ev << getFullPath() << ": Complete Stream\n";
	
	ev<< "Rank " << rank << "\n";
	ev<< "Started at " << startTime << "\n";
	ev<< "Advertise at " << advTime << "\n";
	ev<< "Stable at " << stableTime << "\n";
	recordScalar("Started", startTime);
	recordScalar("Advertise", advTime);
	recordScalar("Stable", stableTime);
	recordScalar("Startup", advTime-startTime);
	recordScalar("FullStartup", stableTime-startTime);
	recordScalar("Rank", rank);
	recordScalar("Packets", numReceived);
	recordScalar("Bytes", byteCount);
		
	ev << "\n----------\n";

	cancelAndDelete(ttimer);
	cancelAndDelete(dtimer);
	UDPBasicApp::finish();
}

//Busca na tabela pelo endereco e retorna o nome do modulo correspondente
std::string PeerApp::getModuleName (std::string addr) {

	for (unsigned i = 0; i < table.size(); i++)
		if (table[i].address == addr)
			return table[i].moduleName;
		
	
	return "";
}

//Retorna a quantidade de saltos ate o endereco
int PeerApp::countHops (std::string a) {
	cTopology::Node *peer = topo.getNodeFor( simulation.getModuleByPath(getModuleName(a).c_str()) );
	return peer->getDistanceToTarget();
}

//Pedido de lista de peers ao tracker
void PeerApp::requestPeerList() {
	PeerRequest *preq;
	preq = new PeerRequest();
	preq->setPeerID(id);
	preq->setStreamID(myStreamID);
	preq->setSrcAddr(s_addr.c_str());
	preq->setDstAddr(s_trackerAddr.c_str());
	preq->setRank(rank);
	ev << getFullPath() << ": Request Peer list to tracker " << preq->getDstAddr() << "\n";
	sendToUDP (preq, localPort, trackerAddr, destPort);
	
	//TESTE
	cTopology::Node *track = topo.getNodeFor( simulation.getModuleByPath(getModuleName(s_trackerAddr).c_str()) );
	ev << getFullPath() << ": Distance to tracker is " << track->getDistanceToTarget() << "\n";
	
}

//Escolhe o proximo peer disponivel
void PeerApp::setNextPeer() {
	while (peerList[peerListIt].isUsed()) {
		peerList[peerListIt].setUse(false);
		
		peerListIt++;
		if (peerListIt >=peerList.size())
			peerListIt = 0;
	}
}

//Retorna o endereco do proximo peer a ser consultado
std::string PeerApp::getNextPeerAddr() {
	if (peerList.size() == 0)
		return "";
	
	std::string ad;
	
	if (peerListIt >= peerList.size())
		peerListIt = 0;
	
	//Usado apenas no modelo geoaware
	if (geo)
		setNextPeer();
	
	ad = peerList[peerListIt].getAddress();

	//Usado apenas no modelo geoaware
	if (geo)
		peerList[peerListIt].setUse(true);
	
	peerListIt++;
	
	return ad;
}

//Retorna o proximo peer a ser consultado
Candidate *PeerApp::getNextPeer() {
	if (peerList.size() == 0)
		return NULL;
	
	Candidate *c;
	
	if (peerListIt >= peerList.size())
		peerListIt = 0;
		
	//Usado apenas no modelo geoaware
	if (geo)
		setNextPeer();
	
	c = &(peerList[peerListIt]);
	
	//Usado apenas no modelo geoaware
	if (geo)
		peerList[peerListIt].setUse(true);
	
	peerListIt++;
	
	return c;
}

//Retorna true se ainda ha dados a serem enviados
void PeerApp::periodicSend() {
	
	bool cont = false;
	IPvXAddress dst;
	Data *data;
	unsigned i = 0;
	
	if (uploads.size() > 0) {
				
		ev << getFullPath() << ": Uploading to " << uploads.size() << " peers at " << simTime() << "\n";
		while (i < uploads.size()) {
			//manda data
			data = new Data();
			data->setPeerID(id);
			data->setStreamID(uploads[i].getStreamID());
			data->setSubStreamID(uploads[i].getSubStreamID());
			data->setSrcAddr(s_addr.c_str());
			data->setDstAddr(uploads[i].getPeerAddr().c_str());
			data->setByteLength(par("messageLength").longValue());
			if (!stable)
				data->setTimestamp(uploads[i].getTimestamp());
			else
				data->setTimestamp(SIMTIME_DBL(simTime()));
			if (dst.tryParse(uploads[i].getPeerAddr().c_str())) {
				sendToUDP (data, localPort, dst, destPort);
				ev << getFullPath() << ": Send data to " << uploads[i].getPeerAddr() << " Stream " << uploads[i].getStreamID()
				<< " Substream " << uploads[i].getSubStreamID() << "\n";
				//Atualiza
				uploads[i].update();
			}
			else {
				ev << getFullPath() << ": IPxAddress Parse error " << uploads[i].getPeerAddr() << "\n";
				delete data;
			}
				
			i++;
		}
			
		cont = true;
		
	}

	if (cont)
		if (!dtimer->isScheduled()) {
			cancelAndDelete(dtimer);
			dtimer = new DataTimer();
			scheduleAt(simTime()+par("sendDataTime"), dtimer);
		}
}

//Retorna o chunk atual do SubStream mais adiantado
int PeerApp::getLastChunk() {
	int max = 0;
	for (unsigned subs = 0; subs < MAX_SUBSTREAM; subs++) {
		if (downloads[subs] != NULL) {
			if (downloads[subs]->getNextChunk() > max)
				max = downloads[subs]->getNextChunk();
		}
	}
	
	return max;
}

//Checa o estado dos substreams e retorna true se houver algum substream nao atendido ou desatualizado
bool PeerApp::downloadCheck() {
	if (stable)
		return false;
	
	unsigned subs;
	bool cont = false;
	
	//Pesquisa qual o substream mais adiantado
	int max = getLastChunk();
	
	//Pesquisa se ha algum substream desatualizado
	for (subs = 0; subs < MAX_SUBSTREAM; subs++) {
		if (downloads[subs] != NULL) {
			//Substream esta offline
			if (!downloads[subs]->isOnline()) {
				cont = true;
				//Imediatamente busca novos peers
				searchPeers(subs);
			}
			else if ( (max - downloads[subs]->getNextChunk()) > (int)par("minChunks") ) {
				//Substream esta desatualizado: diference entre o mais adiantado e ele excede o minimo aceitavel
				closeConnection (downloads[subs]->getPeerID(), downloads[subs]->getStreamID(), downloads[subs]->getSubStreamID(),
					downloads[subs]->getPeerAddr(), EXIT_UPDATE, false, true);
				downloads[subs]->setOffline();
				cont = true;
				//Imediatamente busca novos peers
				searchPeers(subs);
			}
			
		}
		else {
			//Substream ainda nao foi recebido
			cont = true;
			//Imediatamente busca novos peers
			searchPeers(subs);
		}
	}

	ev << getFullPath() << ": Peer Download substream\n";
	ev << "----------------------------------\n";
	for (subs = 0; subs < MAX_SUBSTREAM; subs++) {
		if (downloads[subs] != NULL) {
			if (downloads[subs]->isOnline())
				ev << subs << " " << downloads[subs]->getPeerAddr() << " hops " << countHops(downloads[subs]->getPeerAddr()) << " chunks " << downloads[subs]->getNextChunk()<<"\n";
			else 
				ev << subs << " " << downloads[subs]->getPeerAddr() << " hops " << countHops(downloads[subs]->getPeerAddr()) << " chunks " << downloads[subs]->getNextChunk() <<" Offline\n";
		}
		else
			ev << subs << " xxxxxx\n";
	}
	ev << "----------------------------------\n";

	return cont;

}

void PeerApp::trackingCheck() {
	if (stable)
		return;
	
	bool cont;
	
	//cont = downloadCheck();
	
	if (geo)
		getAllPeerMetrics();
	else
		for (unsigned i = 0; i < peerAddresses.size(); i++)
			peerList.push_back(Candidate(peerAddresses[i], 0, countHops (peerAddresses[i])));
	
	//if ( (peerList.size() < TRACKER_MIN_PEERS) && (cont) ) {
	if (peerList.size() < TRACKER_MIN_PEERS) {
		//Deve pedir uma nova lista ao tracker
		requestPeerList();
		cont = true;
	}
	
	if (cont)
		//Agendando verificacao do estado das requisições
		if (!ttimer->isScheduled()) {
			cancelAndDelete(ttimer);
			ttimer = new TrackTimer();
			scheduleAt(simTime()+par("trackCheckTime"), ttimer);
		}
}

//Busca Peers na rede para fornecer um dado substream, retorna true se a busca foi atendida com sucesso
bool PeerApp::searchPeers(int subs) {

	if (peerList.size() == 0) {
		ev << getFullPath() << ": There is not peers to download\n";
		return false;
	}

	if (subs >= MAX_SUBSTREAM) {
		ev << getFullPath() << ": Invalid SubStream\n";
		return false;
	}

	StreamReq *sreq;
	IPvXAddress dst;
	
	ev << getFullPath() << ": Peer List\n";
	ev << "----------------------------------\n";
	for (unsigned i = 0; i < peerList.size(); i++) {
		if (peerListIt != i)
			ev << peerList[i].getAddress() << " " << peerList[i].getDistance() << " " << peerList[i].isUsed() << "\n";
		else
			ev << peerList[i].getAddress() << " " << peerList[i].getDistance() << " " << peerList[i].isUsed() << "*\n";
	}
	ev << "----------------------------------\n";
	
	int chunk;
	if (downloads[subs] == NULL)
		chunk = 0;
	else
		chunk = downloads[subs]->getNextChunk();

	sreq = new StreamReq();
	sreq->setPeerID(id);
	sreq->setStreamID(myStreamID);
	sreq->setSubStreamID(subs);
	sreq->setSrcAddr(s_addr.c_str());
	sreq->setDstAddr(getNextPeerAddr().c_str());
	sreq->setRank(rank);
	sreq->setChunk(chunk);
	ev << getFullPath() << ": Request to peer " << sreq->getDstAddr() << " substream " << sreq->getSubStreamID() <<"\n";
	if (dst.tryParse(sreq->getDstAddr())) {
		sendToUDP (sreq, localPort, dst, destPort);
	}
	else {
		ev << getFullPath() << ": IPxAddress Parse error " << sreq->getDstAddr() << "\n";
		delete sreq;
		return false;
	}
	
	return true;
}

//Busca Peers na rede para fornecer o proximo substream
bool PeerApp::searchPeers() {
	unsigned s = 0;
	
	while (s < MAX_SUBSTREAM) {
		if (downloads[subStreamCount] == NULL)
			break;
		else if (!downloads[subStreamCount]->isOnline())
			break;
		else {
			subStreamCount = (subStreamCount + 1) % MAX_SUBSTREAM;
			s++;
		}
		
	}
	
	if (s == MAX_SUBSTREAM)
		return false;

	bool ret = searchPeers(subStreamCount);
	subStreamCount = (subStreamCount + 1) % MAX_SUBSTREAM;
	return ret;
}

//Busca peers na rede para fornecer todos os substreams do stream desejado
void PeerApp::fullSearchPeers() {

	unsigned subs;

	for (subs = 0; subs < MAX_SUBSTREAM; subs++) {
		if (downloads[subs] == NULL)
			searchPeers(subs);
		else if (!downloads[subs]->isOnline())
			searchPeers(subs);
	}
	
	//Agendando verificacao do estado das requisições
	if (!ttimer->isScheduled()) {
		cancelAndDelete(ttimer);
		ttimer = new TrackTimer();
		scheduleAt(simTime()+par("trackCheckTime"), ttimer);
	}
	
}

//Envio de Pings para os enderecos para testar Rtt
void PeerApp::getAllPeerMetrics() {
	unsigned i;
	Ping *p;
	IPvXAddress dst;
	
	for (i = 0; i < peerAddresses.size(); i++) {
		if (peerAddresses[i] == s_addr)
			continue;
		p = new Ping();
		p->setSrcAddr(s_addr.c_str());
		p->setDstAddr(peerAddresses[i].c_str());
		ev << getFullPath() << ": Testing peer " << p->getDstAddr() << "\n";
		if (dst.tryParse(p->getDstAddr()))
			sendToUDP (p, localPort, dst, destPort);
		else {
			ev << getFullPath() << ": IPxAddress Parse error " << p->getDstAddr() << "\n";
			delete p;
		}
	}
	
	peerAddresses.clear();
}

//Se for para ele, responde o ping
void PeerApp::processPing(Ping *p) {

	if (p->getDstAddr() == s_addr) {
		IPvXAddress dst;
		Pong *r = new Pong();
		r->setSrcAddr(s_addr.c_str());
		r->setDstAddr(p->getSrcAddr());
		ev << getFullPath() << ": Reply ping to peer " << r->getDstAddr() << "\n";
		if (dst.tryParse(r->getDstAddr()))
			sendToUDP (r, localPort, dst, destPort);
		else {
			ev << getFullPath() << ": IPxAddress Parse error " << r->getDstAddr() << "\n";
			delete r;
		}
		
	}
}

//Processa pong e armazena metricas
void PeerApp::processPong(Pong *p) {
	if (p->getDstAddr() == s_addr) {
		unsigned i,j = 0;
		double rtt = SIMTIME_DBL(simTime()) - SIMTIME_DBL(p->getCreationTime());
		double d = countHops (p->getSrcAddr());
		ev << getFullPath() << ": Metrics of " << p->getSrcAddr() << ": distance " << d <<  " rtt " << rtt <<"\n";
		for (i = 0; i < peerList.size(); i++)
			if (peerList[i].getAddress() == p->getSrcAddr()) {
				peerList[i].setRtt(rtt);
				peerList[i].setDistance(d);
				break;
			}
			else
				if (peerList[i].getMetric() <= Candidate::calcMetric(rtt,d))
					j++;
		
		//Insere na lista, caso nao esteja la
		if ( i >= peerList.size()) {
			if (geo) {
				peerList.insert( peerList.begin()+j, Candidate(p->getSrcAddr(), rtt, d) );
				//Caso o peer inserido seja uma melhor opcao
				if (j < peerListIt)
					peerListIt = j;
			}
			else
				peerList.push_back(Candidate(p->getSrcAddr(), rtt, d));
		}
		
		for (i = 0; i < peerAddresses.size(); i++)
			if (peerAddresses[i] == p->getSrcAddr())
				peerAddresses.erase(peerAddresses.begin()+i);
			else
				i++;
		
		//Deve verificar se ha algum stream a ser transmitido
		if (delayed) {
			if (!dtimer->isScheduled()) {
				cancelAndDelete(dtimer);
				dtimer = new DataTimer();
				scheduleAt(simTime()+par("sendDataTime"), dtimer);
			}
		}
		else
			searchPeers();
	}
	else
		ev << getFullPath() << ": Discard pong from peer " << p->getSrcAddr() << "\n";
}

void PeerApp::processPeerList(PeerList *plist) {

	unsigned i;
	if (plist->getPeerListArraySize() == 0) {
		if (nretries < (int)par("peerListRetries")) {
			//Agendar nova tentativa
			TrackTimer *t = new TrackTimer();
			scheduleAt(simTime()+par("trackCheckTime"), t);
			nretries++;
		}
		return;
	}
	
	nretries = 0;
	std::set<std::string> s;
	for (i = 0; i < peerList.size(); i++) {
		s.insert(peerList[i].getAddress());
	}
	
	for (i = 0; i < plist->getPeerListArraySize(); i++)
		s.insert(plist->getPeerList(i));
	
	peerAddresses.assign(s.begin(), s.end());
	
	if (geo) {
		getAllPeerMetrics();
		//Agendando verificacao do estado das requisições
		if (!ttimer->isScheduled()) {
			cancelAndDelete(ttimer);
			ttimer = new TrackTimer();
			scheduleAt(simTime()+par("trackCheckTime"), ttimer);
		}
	}
	else {
		peerList.clear();
		for (i = 0; i < peerAddresses.size(); i++)
			peerList.push_back(Candidate(peerAddresses[i], 0, countHops (peerAddresses[i])));
		peerListIt = 0;
		peerAddresses.clear();
		//envio de requisicoes para cada peer da lista
		fullSearchPeers();
	}
				
}

void PeerApp::processData(Data *data) {

	IPvXAddress dst;
	
	ev << getFullPath() << ": Received data from " << data->getSrcAddr() << " substream " << data->getSubStreamID() << "\n";
	if (downloads[data->getSubStreamID()] == NULL) { //nao tinha recebido este substream
		//inscreve substream
		downloads[data->getSubStreamID()] = new SubStream(data->getSubStreamID(), data->getStreamID(), data->getPeerID(), data->getSrcAddr(), rank);
		downloads[data->getSubStreamID()]->setTimestamp(data->getTimestamp());
		int max = getLastChunk();
		while (max > 0) {
			max--;
			downloads[data->getSubStreamID()]->update();
		}
		//Agenda verificacao de dados recebidos
		if (!stable)
			if (!dtimer->isScheduled()) {
				cancelAndDelete(dtimer);
				dtimer = new DataTimer();
				scheduleAt(simTime()+par("sendDataTime"), dtimer);
			}
	}
	
	if (!downloads[data->getSubStreamID()]->isOnline()) {
		//Troca de peers
		downloads[data->getSubStreamID()]->changePeer(data->getPeerID(), data->getSrcAddr());
		downloads[data->getSubStreamID()]->setTimestamp(data->getTimestamp());
		int max = getLastChunk() - downloads[data->getSubStreamID()]->getNextChunk();
		while (max > 0) {
			max--;
			downloads[data->getSubStreamID()]->update();
		}
	}
	
	if (downloads[data->getSubStreamID()]->getPeerID() != data->getPeerID()) {
		//Recebeu o dado de outro peer
		closeConnection (data->getPeerID(), data->getStreamID(), data->getSubStreamID(), data->getSrcAddr(), EXIT_ERROR, false, false);
	}
	else {
	//if (downloads[data->getSubStreamID()]->getPeerID() == data->getPeerID()) {
		
		//Recebeu o dado
		if (data->getTimestamp() >= downloads[data->getSubStreamID()]->getTimestamp()) {
			downloads[data->getSubStreamID()]->update();
			downloads[data->getSubStreamID()]->setTimestamp(data->getTimestamp());
			//if (data->getSrcAddr() != s_addr)
			//		byteCount+= ( countHops(data->getSrcAddr())*data->getByteLength() );
		}
		else
			ev << getFullPath() << ": Data is deprecated\n";
		
		for (unsigned i = 0; i < uploads.size(); i++)
			if ( (uploads[i].getStreamID() == data->getStreamID()) and (uploads[i].getSubStreamID() == data->getSubStreamID()) )
				uploads[i].setTimestamp(data->getTimestamp());
		
		bool flag = true;
		
		//Testando condicao de advertisement
		if (advTime == 0)
			for (unsigned j = 0; j < MAX_SUBSTREAM; j++) {
				if (downloads[j] == NULL)
					flag = false;
				else if (downloads[j]->getNextChunk() < (int) par("chunksToAdv"))
					flag = false;
				
				if (!flag)
					break;
			}
		
		if ( (flag) and (advTime == 0)) {
			//mandar advertisement ao tracker
			advTime = simTime();
			Advertisement *padv = new Advertisement();
			padv->setPeerID(id);
			padv->setStreamID(myStreamID);
			padv->setSrcAddr(s_addr.c_str());
			padv->setDstAddr(s_trackerAddr.c_str());
			ev << getFullPath() << ": Advertising tracker " << padv->getDstAddr() << "\n";
			if (dst.tryParse(data->getSrcAddr()))
				sendToUDP (padv, localPort, trackerAddr, destPort);
			else {
				ev << getFullPath() << ": IPxAddress Parse error " << padv->getDstAddr() << "\n";
				delete padv;
			}
		}
		
		//Testando condicao de stable
		if (!stable) {
			flag = true;
			for (unsigned j = 0; j < MAX_SUBSTREAM; j++) {
				if (downloads[j] == NULL)
					flag = false;
				else if (downloads[j]->getNextChunk() <= (int) par("maxChunks"))
					flag = false;
				
				if (!flag)
					break;
			}
		}
		else
			flag = false;
		
		if (flag) {
			stable = true;
			stableTime = simTime();
			if (ev.isGUI())
				getParentModule()->bubble("Stable");
		}
		
		if (stable) {
			//Estabilizado (otimizacao da simulacao)
			std::string srcAddr(data->getSrcAddr());
			closeConnection (data->getPeerID(), data->getStreamID(), data->getSubStreamID(), srcAddr, EXIT_STABLE, false, false);
		}
		
	}
}

//Processa Stream Requests
void PeerApp::processRequest(StreamReq *sreq) {

	IPvXAddress dst;
	Data *data;

	ev << getFullPath() << ": Received request from " << sreq->getSrcAddr() << " substream " << sreq->getSubStreamID() << "\n";

	//envio de dados e registro do peer
	bool serve = false;
	int cause;
	double timestamp = 0;
	
	//verifica se pode servir ao peer
	//Testa a carga
	if (uploads.size() < maxUploads) {
		//Testa se ele possui o substream
		if (downloads[sreq->getSubStreamID()] != NULL) {
			//Testa se possui uma quantidade mínima de chunks
			if ((downloads[sreq->getSubStreamID()]->getNextChunk() - sreq->getChunk()+1) < (int) par("minChunks"))
				cause = EXIT_UPDATE;
			else {
				serve = true;
				timestamp = downloads[sreq->getSubStreamID()]->getTimestamp();
			}
		}
		else
			cause = EXIT_UPDATE; //Nao possui o substream
		
			
	}
	else
		cause = EXIT_LOAD; //esta sobre carga maxima
		
	if (serve) {
		ev << getFullPath() << ": Uploading to " << sreq->getSrcAddr() << "\n";
		data = new Data();
		data->setPeerID(id);
		data->setStreamID(sreq->getStreamID());
		data->setSubStreamID (sreq->getSubStreamID());
		data->setSrcAddr(s_addr.c_str());
		data->setDstAddr(sreq->getSrcAddr());
		data->setByteLength(par("messageLength").longValue());
		
		if (!stable)
			data->setTimestamp(timestamp);
		else
			data->setTimestamp(SIMTIME_DBL(simTime()));
				
		if (dst.tryParse(sreq->getSrcAddr())) {
			sendToUDP (data, localPort, dst, destPort);
			ev << getFullPath() << ": Send data to " << sreq->getSrcAddr() << " Stream " << sreq->getStreamID()
					<< " Substream " << sreq->getSubStreamID() << "\n";
			SubStream *s = new SubStream(sreq->getSubStreamID(), sreq->getStreamID(), sreq->getPeerID(), sreq->getSrcAddr(), sreq->getRank());
			s->update();
			uploads.push_back(*s);
			
			//Agenda o envio de mensagens aos peers clientes
			if (!dtimer->isScheduled()) {
				cancelAndDelete(dtimer);
				dtimer = new DataTimer();
				scheduleAt(simTime()+par("sendDataTime"),dtimer);
			}
		}
		else {
			ev << getFullPath() << ": IPxAddress Parse error " << sreq->getSrcAddr() << "\n";
			delete data;
		}
	}
	else {
		//Nega o pedido ao peer
		std::string s(sreq->getSrcAddr());
		closeConnection (sreq->getPeerID(), sreq->getStreamID(), sreq->getSubStreamID(), s, cause, false, false);
		ev << getFullPath() << ": Will not serve " << sreq->getSrcAddr() << " Peer " << sreq->getPeerID() << "\n";
	}

}

//Envio de pedido de desconexao
void PeerApp::closeConnection (int peerID, int streamID, int subStreamID, std::string addr, int cause, bool upload, bool download) {
	
	IPvXAddress dst;
	PeerExit *pext = new PeerExit();
	pext->setPeerID(id);
	pext->setStreamID(streamID);
	pext->setSubStreamID(subStreamID);
	pext->setSrcAddr(s_addr.c_str());
	pext->setDstAddr(addr.c_str());
	pext->setCause(cause);
	unsigned i = 0;
	
	if (dst.tryParse(pext->getDstAddr())) {
		sendToUDP (pext, localPort, dst, destPort);
	}
	else {
		ev << getFullPath() << ": IPxAddress Parse error " << pext->getDstAddr() << "\n";
		delete pext;
	}
	
	if (upload) {
		i = 0;
		while (i < uploads.size())
			if (uploads[i].compareTo(subStreamID, streamID, peerID))
				uploads.erase(uploads.begin()+i);
			else
				i++;
	}
	
	if (download) {
	
		bool sched = false;
		unsigned subs = subStreamID;
		if ( (subs >= MAX_SUBSTREAM) or (subs < 0) )
			return;
	
		if (downloads[subs] != NULL) {
			if (peerID == downloads[subs]->getPeerID()) {
				if (cause == EXIT_NONE) {
					//O peer saiu da rede, deve remover da lista de peers
					i = 0;
					while (i < peerList.size())
						if(peerList[i].getAddress() == downloads[subs]->getPeerAddr())
							peerList.erase(peerList.begin()+i);
						else
							i++;
				}
			downloads[subs]->setOffline();
			if (!stable)
				sched = true;
			}
		}
		if (sched) {
			//Agendando verificacao
			if (!ttimer->isScheduled()) {
				cancelAndDelete(ttimer);
				ttimer = new TrackTimer();
				scheduleAt(simTime()+par("trackCheckTime"), ttimer);
			}
		}
	}
}

//Recebimento de pedido de desconexao de um peer
void PeerApp::disconnectPeer(PeerExit *pext) {
	
	ev << getFullPath() << ": Received exit from " << pext->getSrcAddr() << " substream " << pext->getSubStreamID() << " cause "<< pext->getCause()<< "\n";
	
	unsigned i = 0;
	bool sched = false;
	
	//if ( (pext->getCause() == EXIT_NONE) or (pext->getCause() == EXIT_STABLE) or (pext->getCause() == EXIT_ERROR))
		while (i < uploads.size())
			if (uploads[i].compareTo(pext->getSubStreamID(), pext->getStreamID(), pext->getPeerID()))
				uploads.erase(uploads.begin()+i);
			else
				i++;
				
	unsigned subs = pext->getSubStreamID();
	if ( (subs >= MAX_SUBSTREAM) or (subs < 0) )
		return;
		
	if (downloads[subs] != NULL) {
		if (pext->getPeerID() == downloads[subs]->getPeerID()) {
			if (pext->getCause() == EXIT_NONE) {
				//O peer saiu da rede, deve remover da lista de peers
				i = 0;
				while (i < peerList.size())
					if(peerList[i].getAddress() == downloads[subs]->getPeerAddr()) 
						peerList.erase(peerList.begin()+i);
					else
						i++;
			}
			
			else if ( (pext->getCause() == EXIT_LOAD) or (pext->getCause() == EXIT_UPDATE)) {
				//O peer ainda pode ser requisitado posteriormente
				i = 0;
				while (i < peerList.size())
					if(peerList[i].getAddress() == downloads[subs]->getPeerAddr()) {
						//Move o peer para o fim da lista
						peerList.erase(peerList.begin()+i);
						peerAddresses.push_back(pext->getSrcAddr());
						break;
					}
					else
						i++;
			}
			
			if (!stable) {
				downloads[subs]->setOffline();
				sched = true;
			}
		}
	}
	
	if (sched) {
		//Agendando verificacao
		if (!ttimer->isScheduled()) {
			cancelAndDelete(ttimer);
			ttimer = new TrackTimer();
			scheduleAt(simTime()+par("trackCheckTime"), ttimer);
		}
	}
}

void PeerApp::handleMessage(cMessage *msg) {
	
	General *gmsg;
	int type;

	if (msg->isSelfMessage()) {
		switch (msg->getKind()) {
			case CONFIGURE:
				startTime = simTime();
				addr = IPAddressResolver().resolve(getParentModule()->getFullName());
				s_addr = addr.str();
				trackerAddr = IPAddressResolver().resolve("tracker");
				s_trackerAddr = trackerAddr.str();
				//Pede lista ao tracker
				requestPeerList();
				delete msg;
				break;
			
			case DATATIMER:
				periodicSend();
				if (!stable) {
					if (downloadCheck())
						//Agendando verificacao do estado das requisições
						if (!ttimer->isScheduled()) {
							cancelAndDelete(ttimer);
							ttimer = new TrackTimer();
							scheduleAt(simTime()+par("trackCheckTime"), ttimer);
						}
					
					if (!dtimer->isScheduled()) {
						cancelAndDelete(dtimer);
						dtimer = new DataTimer();
						scheduleAt(simTime()+par("sendDataTime"), dtimer);
					}
				}
				break;
			
			case TRACKTIMER:
				if (!stable)
					trackingCheck();
			default:
				break;
				
		}
		
		return;
	}
	else {
		//gmsg = (General *) msg;
		if (dynamic_cast<General *> (msg) != NULL) {
			gmsg = (General *) msg;
			type = gmsg->getType();
			numReceived++;
		}
		else
			type = -1;
		if (gmsg != NULL)		
			type = gmsg->getType();
		else
			type = -1;
			
		switch (type) {
			case PING:
				processPing((Ping *) gmsg);
				break;
			case PONG:
				if (!stable)
					processPong((Pong *) gmsg);
				break;
		
			case PEERLIST:
				//Lista de peers
				if(!stable)
					processPeerList((PeerList *) gmsg);
				break;
		
			case STREAMREQ:
				//Pedido de Stream
				processRequest((StreamReq *) gmsg);
				break;
			case PEEREXIT:
				//Desconexão de um peer
				disconnectPeer((PeerExit *) gmsg);
				break;
			case DATA:
				//Dados
				processData((Data *) gmsg);
				if (gmsg->getSrcAddr() != s_addr)
					byteCount+= ( countHops(gmsg->getSrcAddr())*gmsg->getByteLength() );
				break;
			default:
				break;
		}
		
	}
	delete msg;
}
